import glob

from pyspark.sql import *
from pyspark.sql.types import *
import logging
import os
import json
from conf import *


class DataProcess:
    def __init__(self):
        if MODE == Mode.local:
            self.spark = SparkSession.builder.appName("汽车大数据分析").master("local[*]").getOrCreate()
        elif MODE == Mode.distribution:
            self.spark = SparkSession.builder.appName("汽车大数据分析").master('spark://centos1:7077').getOrCreate()
        if DEBUG:
            logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        else:
            logging.basicConfig(level=logging.WARN, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        self.logger = logging.getLogger(__name__)
        self.sc = self.spark.sparkContext
        self.sc.setLogLevel("WARN")

    def clean(self):
        for infile in glob.glob(os.path.join(OUTPUT_PATH, '*.json')):
            os.remove(infile)
        self.logger.info("CLEAN CACHE")

    def analysis(self, num):
        def load(filePath, type='csv'):
            if type == 'csv':
                return self.spark.read.options(header='True', inferScheme='True').csv(filePath)
            return None

        def saveToJson(dataFrame, filePath):
            items = dataFrame.rdd.collect()
            items = [item.asDict() for item in items]
            str = json.dumps(items, ensure_ascii=True)
            with open(filePath, "w") as fOut:
                fOut.write(str)
            return str

        def Task1():
            outputFilePath = OUTPUT_PATH + '//' + 'CarMonthlySales.json'
            inputFilePath = INPUT_PATH + '//' + AREA_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                #df.printSchema()
                if df:
                    df = df.withColumn("number", df["number"].cast(IntegerType()))
                    df = df.groupBy('time').sum('number').sort('time', ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task2():
            outputFilePath = OUTPUT_PATH + '//' + 'CarPriceTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + AREA_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("number", df["number"].cast(IntegerType()))
                    df = df.groupBy('PriceEcharts').sum('number').sort('PriceEcharts', ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task3():
            outputFilePath = OUTPUT_PATH + '//' + 'CarBrandTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('carP_name').sum('car_number').sort('sum(car_number)', ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task4():
            outputFilePath = OUTPUT_PATH + '//' + 'CarTypeMonthlySales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('car_type2', 'sell_time').sum('car_number').sort('car_type2', 'sell_time',
                                                                                     ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task5():
            outputFilePath = OUTPUT_PATH + '//' + 'CarTypeTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('car_type2').sum('car_number')
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task6():
            outputFilePath = OUTPUT_PATH + '//' + 'CarSerialTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('car_name').sum('car_number').sort('sum(car_number)', ascending=False)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task7():
            outputFilePath = OUTPUT_PATH + '//' + 'CarSerialTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('car_name', 'sell_time').sum('car_number').sort('car_name', 'sell_time',
                                                                                    ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task8():
            outputFilePath = OUTPUT_PATH + '//' + 'CarSerialTotalSalesTop.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.groupBy('car_name').sum('car_number').sort('sum(car_number)', ascending=False)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task9():
            outputFilePath = OUTPUT_PATH + '//' + 'CarMPVTotalSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.where("car_type2=='MPV'").groupBy('car_name').sum('car_number').sort('car_name',
                                                                                                 ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        def Task10():
            outputFilePath = OUTPUT_PATH + '//' + 'CarMPVMonthSales.json'
            inputFilePath = INPUT_PATH + '//' + SERIAL_SALES
            if os.path.exists(outputFilePath):
                self.logger.info('LOAD DATA_SET')
                with open(outputFilePath, 'r') as f:
                    result = json.load(f)
                return result
            else:
                self.logger.info('GENERATE DATA_SET')
                df = load(inputFilePath)
                if df:
                    df = df.withColumn("car_number", df["car_number"].cast(IntegerType()))
                    df = df.where("car_type2=='MPV'").groupBy('car_name', 'sell_time').sum('car_number').sort(
                        'car_name', 'sell_time', ascending=True)
                    return saveToJson(df, outputFilePath)
                self.logger.warning("NOT GET")
                return None

        self.logger.info(f'SELECT {num} DATA_SET')
        jsonStr = ''
        if num == 1:
            jsonStr = Task1()
        elif num == 2:
            jsonStr = Task2()
        elif num == 3:
            jsonStr = Task3()
        elif num == 4:
            jsonStr = Task4()
        elif num == 5:
            jsonStr = Task5()
        elif num == 6:
            jsonStr = Task6()
        elif num == 7:
            jsonStr = Task7()
        elif num == 8:
            jsonStr = Task8()
        elif num == 9:
            jsonStr = Task9()
        elif num == 10:
            jsonStr = Task10()
        return jsonStr
